package com.xxxx.keyTest;


import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @program: flink19Test
 * @description: tuple处理key
 * @author: CoreDao
 * @create: 2021-04-01 11:05
 **/

public class TupleKeyTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        DataStreamSource<String> textStream = env.socketTextStream("localhost", 9998);
        //Tuple3
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = textStream.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String value) throws Exception {
                String[] split = value.split(" ");
                return Tuple3.of(split[0], split[1], Integer.parseInt(split[2]));
            }
        });
        //查看KeyBy操作的对象，针对的元素是第二个
        KeyedStream<Tuple3<String, String, Integer>, Tuple> keyedStream = map.keyBy(0);
        KeyedStream<Tuple3<String, String, Integer>, Tuple> keyedStream1 = map.keyBy(1);
        KeyedStream<Tuple3<String, String, Integer>, Tuple> keyedStream2 = map.keyBy(0,1);
        KeyedStream<Tuple3<String, String, Integer>, Tuple> keyedStream4 = map.keyBy("f0");
        KeyedStream<Tuple3<String, String, Integer>, String> KeyedStream5 = map.keyBy(tuple -> tuple.f1);

        //上面这些都是针对Tuple这个特殊类，可以写的指定Key的特殊方式。
        keyedStream.reduce(new ReduceFunction<Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> value1, Tuple3<String, String, Integer> value2) throws Exception {
                return Tuple3.of(value1.f0 + "f0" + value2.f0,value1.f1 + "f1" + value2.f1,value1.f2 + value2.f2);
            }
        }).print();



        env.execute();
    }
}
